package org.stratusdb.funsql.execute;

import java.util.HashMap;

import org.stratusdb.Config;
import org.stratusdb.error.Error;
import org.stratusdb.types.TableType;
import org.stratusdb.types.TupleType;
import org.stratusdb.types.AbstractSimpleType;

public class HashEquiJoin extends BinaryOperator {

	private static final long serialVersionUID = 3374921378192699529L;

	// join attributes
	private int leftAtt;
	private int rightAtt;
	private TableType leftTable;
	private int leftCount;
	private TableType rightTable;
	private int rightCount;

	// hash table
	private HashMap<AbstractSimpleType, TableType> rightMap = new HashMap<AbstractSimpleType, TableType>();

	// constructors
	public HashEquiJoin(AbstractOperator leftChild,
			AbstractOperator rightChild, int leftAtt, int rightAtt) {
		super(leftChild, rightChild);

		this.leftAtt = leftAtt;
		this.rightAtt = rightAtt;
	}

	// methods
	@Override
	public Error open() {
		// scan right child
		this.lastError = this.rightChild.open();
		if (lastError != Error.NO_ERROR)
			return this.lastError;

		this.rightTable = null;
		while ((this.rightTable = this.rightChild.next()) != null) {
			for (TupleType rightTuple : this.rightTable.getTuples()) {
				this.lastError = rightChild.getLastError();
				if (this.lastError != Error.NO_ERROR)
					return this.lastError;

				// add tuple to map
				AbstractSimpleType rightKey = rightTuple
						.getValue(this.rightAtt);
				TableType rightValue;
				if (!this.rightMap.containsKey(rightKey)) {
					rightValue = new TableType();
					this.rightMap.put(rightKey, rightValue);
				} else {
					rightValue = this.rightMap.get(rightKey);
				}
				rightValue.addTuple(rightTuple);
			}
		}
		this.rightTable = null;
		this.rightCount = 0;

		this.lastError = this.rightChild.close();
		if (this.lastError != Error.NO_ERROR)
			return this.lastError;

		// open left child
		this.lastError = this.leftChild.open();
		if (this.lastError != Error.NO_ERROR)
			return this.lastError;

		this.leftTable = this.leftChild.next();
		this.leftCount = 0;
		this.lastError = this.leftChild.getLastError();

		return this.lastError;
	}

	@Override
	public TableType next() {
		TableType resultTable = new TableType();

		loop1: while (resultTable.size() < Config.EXECUTE_MAX_FETCHSIZE
				&& this.leftTable != null) {

			// continue with current table from left child
			if (this.leftCount < this.leftTable.size()) {

				// get current left tuple with current position
				TupleType leftTuple = this.leftTable.getTuple(this.leftCount);

				// get next table from right child
				if (this.rightTable == null) {
					AbstractSimpleType leftKey = leftTuple
							.getValue(this.leftAtt);
					this.rightTable = this.rightMap.get(leftKey);
					this.rightCount = 0;

					this.lastError = this.rightChild.getLastError();
					if (this.lastError != Error.NO_ERROR)
						return null;
				}

				// join left tuple with current position in right table
				while (rightTable != null
						&& this.rightCount < this.rightTable.size()) {
					TupleType rightTuple = this.rightTable
							.getTuple(this.rightCount);
					resultTable.addTuple(leftTuple.clone().append(
							rightTuple.clone()));
					this.rightCount++;

					if (resultTable.size() == Config.EXECUTE_MAX_FETCHSIZE) {
						break loop1;
					}
				}

				// end of right table reached -> new left tuple
				this.rightTable = null;
				this.rightCount = 0;
				this.leftCount++;
			}
			// fetch new table from left child
			else {
				this.leftTable = this.leftChild.next();
				this.leftCount = 0;

				this.lastError = this.leftChild.getLastError();
				if (this.lastError != Error.NO_ERROR)
					return null;
			}
		}

		// return result
		if (resultTable.size() == 0)
			return null;
		else
			return resultTable;
	}

	@Override
	public Error close() {
		// clean up
		this.rightMap.clear();
		this.rightCount = 0;
		this.rightTable = null;
		this.leftCount = 0;
		this.leftTable = null;

		// close left child
		this.lastError = this.leftChild.close();
		return this.lastError;
	}

}
